package com.etlmaster.executor.steps.table;

import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.client.HBaseAdmin;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.bigdata.hbase.HbaseUtil;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.jdbc.JdbcTemplate;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("crtHbaseTab")
public class CreateHbaseTableStep extends Step {

	private CreateHbaseTableStepMeta meta;

	public CreateHbaseTableStep(StepMeta meta) {
		super(meta);
		this.meta = (CreateHbaseTableStepMeta) meta;
	}

	@Override
	protected int handle() {
		HBaseAdmin admin=null;
		try {
			JdbcTemplate jdbc = new JdbcTemplate();
			String fetchColumnInfo = DataCollectionUtils
					.getProperties("fetchHbaseColumnInfo");
			List<Map<String, Object>> list = jdbc.queryForList(fetchColumnInfo,
					new Object[] { meta.getXmlId() });
			if (list != null && list.size() > 0) {
				String[] columns = new String[list.size()];
				for (int i = 0; i < list.size(); i++) {
					columns[i] = (String) list.get(i).get("colname");
				}
				admin=new HBaseAdmin(HbaseUtil.getConf());
				HbaseUtil.create(admin,meta.getTableName(), columns,
						Boolean.valueOf(meta.getOverwrite()));
				
				return StepState.RUNNING_SUCCESSED;
			} else {
				meta.log("ERROR","未找到模型{}相关的字段信息！！",meta.getXmlId());
				return StepState.RUNNING_FAILED;
			}
		} catch (Exception e) {
			meta.log("ERROR", "创建Hbase表{}失败！{}", meta.getTableName(),
					e.getMessage());
			e.printStackTrace();
			return StepState.RUNNING_FAILED;
		}finally{
			HbaseUtil.closeAdmin(admin);
		}

	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
